package com.hu.flink12.api.source;

import com.hu.flink12.api.entity.Order;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @Author: hujianjun
 * @Date: 2021/2/3 22:45
 * @Describe: 自定义订单数据
 *  每隔一秒钟生成一条订单数据
 *  要求：随机生成订单ID
 *       随机生成用户ID(0-2)
 *       随机生成订单金额（0-100）、
 *       时间戳为系统时间
 */
public class CustomerOrder {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 2.读取数据源
        DataStream<Order> orderDataStream = env.addSource(new CustomerOrderSource());

        //TODO 3.转换操作

        //TODO 4.sink操作
        orderDataStream.print();

        //TODO 5.执行程序
        env.execute();
    }

    public static class CustomerOrderSource extends RichParallelSourceFunction<Order>{

        private Boolean flag = true;

        @Override
        public void run(SourceContext<Order> ctx) throws Exception {
            Random random = new Random();
            while (flag){
//                Thread.sleep(5000);
                TimeUnit.SECONDS.sleep(5);
                ctx.collect(new Order(UUID.randomUUID().toString(),random.nextInt(3),random.nextInt(101),System.currentTimeMillis()));
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
